package cn.ucox.web.ms.plugin;

import cn.ucox.web.ms.Constant;
import cn.ucox.web.ms.config.DataExchangeConfig.PollingConfig;
import cn.ucox.web.ms.polling.DataBucket;
import cn.ucox.web.ms.polling.IDataLoader;
import cn.ucox.web.ms.polling.SyncAnchor;
import cn.ucox.web.ms.polling.SyncData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

/**
 * Oracle数据库加载器
 *
 * @author chenw
 * @create 2017-05-11 14:22
 * @email javacspring@gmail.com
 */
public class OracleDataLoader implements IDataLoader {

    private static final Logger logger = LoggerFactory.getLogger(IDataLoader.class);

    private static final String SQL = "SELECT TAB,ACT,CTM,ATT,DATA,SYN_ID FROM ( SELECT ROWNUM AS RN,V.TAB,V.ACT,V.CTM,V.ATT,V.DATA,V.SYN_ID FROM ( SELECT TAB,ACT,CTM,ATT,DATA,SYN_ID FROM OS_DEX_SYNC WHERE CTM BETWEEN ? AND SYSTIMESTAMP ORDER BY CTM ASC) V WHERE ROWNUM <= ? ) WHERE RN > ?";
    private static final String MEDIUMSQL = "SELECT TAB,ACT,CTM,ATT,DATA,SYN_ID FROM ( SELECT ROWNUM AS RN,V.TAB,V.ACT,V.CTM,V.ATT,V.DATA,V.SYN_ID FROM ( SELECT TAB,ACT,CTM,ATT,DATA,SYN_ID FROM OS_DEX_MEDIUM_SYNC WHERE CTM BETWEEN ? AND SYSTIMESTAMP ORDER BY CTM ASC) V WHERE ROWNUM <= ? ) WHERE RN > ?";
    private static final String LOWSQL = "SELECT TAB,ACT,CTM,ATT,DATA,SYN_ID FROM ( SELECT ROWNUM AS RN,V.TAB,V.ACT,V.CTM,V.ATT,V.DATA,V.SYN_ID FROM ( SELECT TAB,ACT,CTM,ATT,DATA,SYN_ID FROM OS_DEX_LOW_SYNC WHERE CTM BETWEEN ? AND SYSTIMESTAMP ORDER BY CTM ASC) V WHERE ROWNUM <= ? ) WHERE RN > ?";

    private static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();

    private PollingConfig mPollingConfig;
    private JdbcTemplate mJdbcTemplate;

    public OracleDataLoader(PollingConfig pollingConfig, JdbcTemplate jdbcTemplate) {
        this.mPollingConfig = pollingConfig;
        this.mJdbcTemplate = jdbcTemplate;
    }

    @Override
    public DataBucket getData(SyncAnchor anchor, String level) {
        long taskStart = System.currentTimeMillis();
        boolean firstTime = (100000000 == anchor.getTransactionId());
        long maxLoadNum = AVAILABLE_PROCESSORS * mPollingConfig.getProcessNumPerCpuThread();
        Timestamp startPoint;

        List<SyncData> result = new ArrayList<>();

        if (level.equalsIgnoreCase(Constant.SYNC_LEVEL_HIGH)) {
            if (firstTime) {
                startPoint = new Timestamp(0);
            } else {
                String[] startPointStr = anchor.getDataEndTime().split("\\.");
                startPoint = new Timestamp(Long.valueOf(startPointStr[0]));
                startPoint.setNanos(Integer.valueOf(startPointStr[1]));
            }
            result = mJdbcTemplate.query(SQL,
                    ps -> {
                        ps.setTimestamp(1, startPoint);
                        ps.setLong(2, firstTime ? maxLoadNum : (maxLoadNum + 1));
                        ps.setInt(3, firstTime ? 0 : 1);
                    },
                    (rs, rowNum) -> {
                        Timestamp time = rs.getTimestamp(3);
                        return new SyncData(
                                rs.getString(1),
                                rs.getInt(2),
                                String.format("%d.%d", (time.getTime() - time.getNanos() / 1000000), (time.getNanos())),
                                rs.getInt(4),
                                clobToString(rs.getCharacterStream(5)),
                                rs.getLong(6));
                    });


        } else if (level.equalsIgnoreCase(Constant.SYNC_LEVEL_MEDIUM)) {
            if (firstTime) {
                startPoint = new Timestamp(0);
            } else {
                String[] startMediumPointStr = anchor.getDataMediumEndTime().split("\\.");
                startPoint = new Timestamp(Long.valueOf(startMediumPointStr[0]));
                startPoint.setNanos(Integer.valueOf(startMediumPointStr[1]));
            }

            result = mJdbcTemplate.query(MEDIUMSQL,
                    ps -> {
                        ps.setTimestamp(1, startPoint);
                        ps.setLong(2, firstTime ? maxLoadNum : (maxLoadNum + 1));
                        ps.setInt(3, firstTime ? 0 : 1);
                    },
                    (rs, rowNum) -> {
                        Timestamp time = rs.getTimestamp(3);
                        return new SyncData(
                                rs.getString(1),
                                rs.getInt(2),
                                String.format("%d.%d", (time.getTime() - time.getNanos() / 1000000), (time.getNanos())),
                                rs.getInt(4),
                                clobToString(rs.getCharacterStream(5)),
                                rs.getLong(6));
                    });

        } else {
            if (firstTime) {
                startPoint = new Timestamp(0);
            } else {
                String[] startLowPointStr = anchor.getDataLowEndTime().split("\\.");
                startPoint = new Timestamp(Long.valueOf(startLowPointStr[0]));
                startPoint.setNanos(Integer.valueOf(startLowPointStr[1]));
            }

            result = mJdbcTemplate.query(LOWSQL,
                    ps -> {
                        ps.setTimestamp(1, startPoint);
                        ps.setLong(2, firstTime ? maxLoadNum : (maxLoadNum + 1));
                        ps.setInt(3, firstTime ? 0 : 1);
                    },
                    (rs, rowNum) -> {
                        Timestamp time = rs.getTimestamp(3);
                        return new SyncData(
                                rs.getString(1),
                                rs.getInt(2),
                                String.format("%d.%d", (time.getTime() - time.getNanos() / 1000000), (time.getNanos())),
                                rs.getInt(4),
                                clobToString(rs.getCharacterStream(5)),
                                rs.getLong(6));
                    });

        }

        //无数据
        if (null == result || 0 == result.size()) {
            logger.debug("增量加载优先级为:{},数据区间:[{}---{}]，限定最大加载{}条,实际加载0条,耗时:{}ms",
                    Constant.SYNC_LEVEL_MAP.get(level),
                    startPoint.toString(),
                    new Timestamp(taskStart).toString(),
                    maxLoadNum,
                    (System.currentTimeMillis() - taskStart));
            return null;
        } else {
            DataBucket bucket = new DataBucket();
            bucket.setSyncData(result);
            bucket.setSyncTime(System.currentTimeMillis());
            String[] startTimeValue = result.get(0).getTime().split("\\.");
            Timestamp start = new Timestamp(Long.valueOf(startTimeValue[0]));
            start.setNanos(Integer.valueOf(startTimeValue[1]));

            String[] endTimeValue = result.get(result.size() - 1).getTime().split("\\.");
            Timestamp end = new Timestamp(Long.valueOf(endTimeValue[0]));
            end.setNanos(Integer.valueOf(endTimeValue[1]));
            if (logger.isDebugEnabled()) {
                logger.debug("增量加载优先级为:{},数据区间:[{}---{}]，限定最大加载{}条,实际加载{}条,实际加载数据时间区间:[{}---{}] 耗时:{}ms",
                        Constant.SYNC_LEVEL_MAP.get(level),
                        startPoint.toString(),
                        new Timestamp(taskStart).toString(),
                        maxLoadNum,
                        result.size(),
                        start,
                        end,
                        (System.currentTimeMillis() - taskStart));
            }
            bucket.setSyncStartTime(startPoint.getTime());
            bucket.setSyncEndTime(new Timestamp(taskStart).getTime());
            return bucket;
        }

    }

    private String clobToString(Reader reader) {
        BufferedReader br = new BufferedReader(reader);
        try {
            String s = br.readLine();
            StringBuilder sb = new StringBuilder();
            while (s != null) {
                sb.append(s);
                s = br.readLine();
            }
            return sb.toString();
        } catch (IOException e) {
            e.printStackTrace();
            return "";
        } finally {
            try {
                br.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    public JdbcTemplate getmJdbcTemplate() {
        return mJdbcTemplate;
    }

    public void setmJdbcTemplate(JdbcTemplate mJdbcTemplate) {
        this.mJdbcTemplate = mJdbcTemplate;
    }
}
